package com.bbx.flink.demo.transformations;

//import com.bbx.flink.demo.entity.Student;
//import com.sun.deploy.util.StringUtils;
//import jdk.internal.util.xml.impl.Input;
import lombok.extern.slf4j.Slf4j;
//import org.apache.flink.api.common.JobExecutionResult;
//import org.apache.flink.api.common.functions.*;
//import org.apache.flink.api.common.functions.util.ListCollector;
//import org.apache.flink.api.common.serialization.SerializationSchema;
//import org.apache.flink.api.common.serialization.SimpleStringSchema;
//import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
//import org.apache.flink.api.common.typeinfo.TypeInformation;
//import org.apache.flink.api.java.functions.KeySelector;
//import org.apache.flink.shaded.akka.org.jboss.netty.util.internal.StringUtil;
//import org.apache.flink.shaded.guava18.com.google.common.primitives.Longs;
//import org.apache.flink.streaming.api.TimeCharacteristic;
//import org.apache.flink.streaming.api.collector.selector.OutputSelector;
//import org.apache.flink.streaming.api.datastream.*;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.sink.SinkFunction;
//import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
//import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
//import org.apache.flink.util.Collector;
//import org.apache.flink.util.OutputTag;
//
//import java.util.ArrayList;
//import java.util.Arrays;
//import java.util.List;
//import java.util.stream.Collectors;


/**
 *
 */
@Slf4j
public class DataStreamTransformationsDemo {

//    /**
//     * map(MapFunction<T, R> mapper)
//     * 通过一个元素产生一个新的元素
//     * @param args
//     * @throws Exception
//     */
//    public static void main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStreamSource<Integer> integerDataStreamSource = env.fromElements(1, 3, 5);
//        SingleOutputStreamOperator<Integer> dataStream = integerDataStreamSource.map(new MapFunction<Integer, Integer>() {
//            private static final long serialVersionUID = -6061817441663213249L;
//
//            @Override
//            public Integer map(Integer value) throws Exception {
//                value = value * 2;
//                return value;
//            }
//        });
//        dataStream.print();
//        env.execute();
//
//    }


//    /**
//     * flatMap(FlatMapFunction<T, R> flatMapper)
//     * 扁平化流，通过一个元素产生零个，一个或者多个元素
//     * @param args
//     * @throws Exception
//     */
//    public static void main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStreamSource<String> stringDataStreamSource = env.fromElements("a b c d", "aABCD", "1 2 3");
//        ArrayList<String> outList = new ArrayList<>();
//        SingleOutputStreamOperator<String> dataStream = stringDataStreamSource.flatMap(new FlatMapFunction<String, String>() {
//            private static final long serialVersionUID = -3782689816919562881L;
//
//            @Override
//            public void flatMap(String value, Collector<String> out) throws Exception {
//                String[] s = value.split(" ");
//                Arrays.stream(s).forEach(i -> out.collect(i));
//
//            }
//        });
//        dataStream.print();
//        env.execute();
//    }


//    /**
//     * filter(FilterFunction<T> filter)
//     *流数据过滤，对每一个元素进行boolean运算，只保留结果为true 的元素
//     * @param args
//     */
//    public static void  main(String [] args) throws Exception {
//        //创建执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        //获取数据源，并且对数据源进行转换过滤
//        SingleOutputStreamOperator<Integer> outputStreamOperator = env.fromElements(1, 5, 20, 40, 35, 27, 33)
//                .filter(new FilterFunction<Integer>() {
//                    private static final long serialVersionUID = 4596950454812092153L;
//
//                    @Override
//                    public boolean filter(Integer value) throws Exception {
//                        return value > 25;
//                    }
//                }
//        );
//        //结果输出
//        outputStreamOperator.print();
//        //触发执行
//        env.execute();
//    }


//    /**
//     * reduce(ReduceFunction<T> reducer)   对KeyedStream 按照key 进行滚动的聚合（压缩）
//     * @param args
//     * @throws Exception
//     */
//    public  static void  main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        List<Student> list = new ArrayList<>();
//        list.add(new Student("1","a",22,"man"));
//        list.add(new Student("2","b",33,"man"));
//        list.add(new Student("3","c",26,"woman"));
//        list.add(new Student("4","d",32,"man"));
//        list.add(new Student("5","e",18,"man"));
//        list.add(new Student("6","f",20,"woman"));
//        SingleOutputStreamOperator<Student> reduce = env.fromCollection(list)
//                .keyBy(i -> i.getGender())
//                .reduce(new ReduceFunction<Student>() {
//                    private static final long serialVersionUID = -7843116673365021624L;
//
//                    @Override
//                    public Student reduce(Student value1, Student value2) throws Exception {
//                        log.info("value1[{}]",value1);
//                        log.info("value2[{}]",value2);
//                        log.info("[{}]","~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
//
//
//
//                        //对分组内的数据进行 年龄比较，返回年龄大的
//                        return value1.getAge()>value2.getAge()?value1:value2;
//                    }
//                });
//        reduce.print();
//        env.execute();
//    }


//    /**
//     * fold(R initialValue, FoldFunction<T, R> folder)
//     * 对KeyedStream 数据流增加初始值然后进行折叠
//     * 大白话：对数据流按照keyBy进行分区，对分区后的数据增加初始值，实在说不清楚了，看下面官方说明，自行体会
//     *
//     * A fold function that, when applied on the sequence (1,2,3,4,5),
//     * emits the sequence "start-1", "start-1-2", "start-1-2-3", ...
//     * @param args
//     * @throws Exception
//     */
//    public static void  main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        SingleOutputStreamOperator<String> operator = env.generateSequence(10L, 20L).keyBy(new KeySelector<Long, String>() {
//
//            private static final long serialVersionUID = 7536704776032117487L;
//
//            @Override
//            public String getKey(Long value) throws Exception {
//                return value % 2 == 0 ? "偶数" : "奇数";
//            }
//        }).fold("start", new FoldFunction<Long, String>() {
//            private static final long serialVersionUID = 2873382577990478083L;
//
//            @Override
//            public String fold(String accumulator, Long value) throws Exception {
//
//                return accumulator + "------" + value;
//            }
//        });
//        operator.print();
//        env.execute();
//    }


//    /**
//     * 聚合函数
//     * 滚动的对KeyedStream 进行聚合，
//     * The difference between min and minBy is that min returns the minimum value, whereas minBy returns the element that has the minimum value in this field (same for max and maxBy).
//     * min ：返回的是最小的值
//     * minBy ：返回的是具有最小值的元素
//     * @param args
//     * @throws Exception
//     */
//    public static void main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        KeyedStream<Long, String> keyedStream = env.generateSequence(1L, 10L).keyBy(new KeySelector<Long, String>() {
//            private static final long serialVersionUID = 2304749345114557909L;
//
//            @Override
//            public String getKey(Long value) throws Exception {
//
//                return value % 2 == 0 ? "偶数" : "奇数";
//            }
//        });
//        SingleOutputStreamOperator<Long> sum = keyedStream.sum(0);
//        SingleOutputStreamOperator<Long> sumEven = keyedStream.sum("*");
//        SingleOutputStreamOperator<Long> max = keyedStream.max(0);
//        sum.print();
//        log.info("[{}]","~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~sum(0)~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
//        sumEven.print();
//        log.info("[{}]","~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~sum(*)~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
//        max.print();
//        log.info("[{}]","~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~max(0)~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
//        env.execute();
//    }

//    /**
//     * timeWindow(Time size)
//     * Windows this {@code KeyedStream} into tumbling time windows.   滚动时间窗口
//     *通过已经分区的keyedStream 创建滚动窗口（tumbling time windows）
//     * @param args
//     * @throws Exception
//     */
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStreamSource<String> stringDataStreamSource = env.socketTextStream("192.168.10.131", 10002);
//
//        stringDataStreamSource
//                .map(new MapFunction<String, Integer>() {
//                    private static final long serialVersionUID = 1723130212830899292L;
//
//                    @Override
//                    public Integer map(String value) throws Exception {
//                        return Integer.parseInt(value);
//                    }
//                })
//                .keyBy(new KeySelector<Integer, String>() {
//                    private static final long serialVersionUID = -4689187447705594609L;
//
//                    @Override
//                    public String getKey(Integer value) throws Exception {
//                        return value % 2 == 0 ? "偶数" : "奇数";
//                    }
//                }).timeWindow(Time.seconds(5L))
//                .max(0)
//                .print();
//        env.execute();
//    }


//    /**
//     *  timeWindowAll(Time size)
//     *  Windows this {@code DataStream} into tumbling time windows.   滚动窗口函数
//     * 通过DataStream 创建窗口，窗口根据某些特性对流进行分组
//     *
//     * @param args
//     * @throws Exception
//     */
//    public  static void  main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.socketTextStream("192.168.10.131",10002)
//                .map(new MapFunction<String, Integer>() {
//
//                    private static final long serialVersionUID = 1723130212830899292L;
//
//                    @Override
//                    public Integer map(String value) throws Exception {
//                        return Integer.parseInt(value);
//                    }
//                })
//                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5L)))
////                .timeWindowAll(Time.seconds(5L))
//                .min(0)
//                .print();
//
//        env.execute();
//    }

//
//    /**
//     * apply(AllWindowFunction<T, R, W> function)
//     * 对窗口应用一个自定义的函数，（自己来实现窗口内元素的操作）
//     * 自定义实现窗口元素求和
//     * @param args
//     * @throws Exception
//     */
//    public  static void  main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.socketTextStream("192.168.10.131",10002)
//                .map(new MapFunction<String, Integer>() {
//
//                    private static final long serialVersionUID = 1723130212830899292L;
//
//                    @Override
//                    public Integer map(String value) throws Exception {
//                        return Integer.parseInt(value);
//                    }
//                })
//                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5L)))
//                .apply(new AllWindowFunction<Integer, Integer, TimeWindow>() {
//                    private static final long serialVersionUID = 4852460822322482580L;
//
//                    @Override
//                    public void apply(TimeWindow window, Iterable<Integer> values, Collector<Integer> out) throws Exception {
//                         int sum = 0;
//                        for(Integer v :values){
//                            sum +=v;
//                        }
//                        out.collect(sum);
//                    }
//                }).print();
//        env.execute();
//    }


//    /**
//     * reduce(ReduceFunction<T> function)
//     * 对窗口应用reduce函数，
//     * @param args
//     * @throws Exception
//     */
//    public  static void  main(String [] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.socketTextStream("192.168.10.131",10002)
//                .map(new MapFunction<String, Integer>() {
//
//                    private static final long serialVersionUID = 1723130212830899292L;
//
//                    @Override
//                    public Integer map(String value) throws Exception {
//                        return Integer.parseInt(value);
//                    }
//                })
//                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5L)))
//                .reduce(new ReduceFunction<Integer>() {
//                    private static final long serialVersionUID = 4565484197133507783L;
//
//                    @Override
//                    public Integer reduce(Integer value1, Integer value2) throws Exception {
//                        return value1+value2;
//                    }
//                }).print();
//        env.execute();
//    }


//    /**
//     *  union(DataStream<T>... streams)
//     *  合并两个或者更多的DataStream,生成一个新的DataStream，这个新的DataStream 包含有所有流中的元素
//     *
//     * @param args
//     * @throws Exception
//     */
//    public static void   main(String []  args ) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStreamSource<Integer> dataStreamSource = env.fromElements(1, 2, 3, 4, 5);
//        DataStreamSource<Integer> dataStreamSource1 = env.fromElements(6, 7, 8, 9, 10);
//        dataStreamSource.union(dataStreamSource1).print();
//        env.execute();
//    }


//*
//     * Iterations   --暂未研究明白
//     * @param args
//     * @throws Exception
//
//
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        SplitStream<Integer> splitStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9)
//                .split(new OutputSelector<Integer>() {
//                    private static final long serialVersionUID = -6537482987418042973L;
//                    List<String> list = new ArrayList<>();
//
//                    @Override
//                    public Iterable<String> select(Integer value) {
//
//                        if (value % 2 == 0) {
//                            list.add("even");
//                        } else {
//                            list.add("odd");
//                        }
//                        return list;
//                    }
//                });
////       splitStream.select("even").print();
//       log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
//       splitStream.select("odd").print();
//        env.execute();
//    }


//    /**
//     *  writeToSocket(String hostName, int port, SerializationSchema<T> schema)
//     *  将流中元素输出到socket，输出格式由  SerializationSchema 决定
//     * @param args
//     * @throws Exception
//     */
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.fromElements("a","B","C","d","e","f")
//                .writeToSocket("192.168.10.131", 10002,new SimpleStringSchema());
//       env.execute();
//    }


//    /**
//     * addSink(SinkFunction<T> sinkFunction)   调用一个自定义的sink 函数，
//     * @param args
//     * @throws Exception
//     */
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.fromElements("a","B","C","d").addSink(new SinkFunction<String>() {
//            private static final long serialVersionUID = -3294535271683312351L;
//
//            @Override
//            public void invoke(String value, Context context) throws Exception {
//                log.info("value~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~[{}]",value);
//            }
//        });
//        env.execute();
//    }

//    /**
//     *
//     * @param args
//     * @throws Exception
//     */
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        IterativeStream<Long> iterate = env.generateSequence(-10L, 10L).iterate();
//
//
//        SingleOutputStreamOperator<Long> minusOne = iterate.map(new MapFunction<Long, Long>() {
//            private static final long serialVersionUID = 9221553395996171401L;
//
//            @Override
//            public Long map(Long value) throws Exception {
//                return value - 1;
//            }
//        });
//        iterate.closeWith(minusOne.filter(new FilterFunction<Long>() {
//            private static final long serialVersionUID = 3834298011326094397L;
//            @Override
//            public boolean filter(Long value) throws Exception {
//                return value>0;
//            }
//        }));
//        env.execute();
//    }



}
